package com.hujifang.elasticsearch.service.impl;


import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.hujifang.elasticsearch.model.EsAggregation;
import com.hujifang.elasticsearch.model.EsDateHistogram;
import com.hujifang.elasticsearch.model.EsDocument;
import com.hujifang.elasticsearch.model.EsField;
import com.hujifang.elasticsearch.model.EsHighlightField;
import com.hujifang.elasticsearch.model.EsIndex;
import com.hujifang.elasticsearch.model.EsJDBCRiverConfig;
import com.hujifang.elasticsearch.model.EsSearch;
import com.hujifang.elasticsearch.model.EsSearchHit;
import com.hujifang.elasticsearch.model.EsSearchResult;
import com.hujifang.elasticsearch.model.EsType;
import com.hujifang.elasticsearch.moudle.Result;
import com.hujifang.elasticsearch.service.ElasticSearchService;
import com.hujifang.elasticsearch.service.exception.ElasticException;
import com.hujifang.elasticsearch.service.utils.SearchConditionUtil;
import com.hujifang.elasticsearch.utils.DateUtil;
import com.hujifang.elasticsearch.utils.HttpUtil;
import com.hujifang.elasticsearch.utils.JSONUtils;
import com.hujifang.elasticsearch.utils.MapUtil;
import com.hujifang.elasticsearch.utils.Pinyin4jUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequestBuilder;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsResponse;
import org.elasticsearch.action.admin.indices.mapping.delete.DeleteMappingRequestBuilder;
import org.elasticsearch.action.admin.indices.mapping.delete.DeleteMappingResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.suggest.SuggestResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.BoolFilterBuilder;
import org.elasticsearch.index.query.FilterBuilder;
import org.elasticsearch.index.query.FilterBuilders;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.index.query.RangeFilterBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation;
import org.elasticsearch.search.aggregations.bucket.SingleBucketAggregation;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogram;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramBuilder;
import org.elasticsearch.search.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TimeZone;
import java.util.concurrent.ConcurrentHashMap;


/**
 * ElasticSearch客户端工具<br/>
 * Created by Hurricane.Hu on 2015/8/4.
 */
public class ElasticSearchServiceImpl implements ElasticSearchService, DisposableBean, InitializingBean {

    private static Logger logger = LoggerFactory.getLogger("#ElasticSearchLog#");

    /**
     * 主机地址，格式是 ip:port,ip1:port1.....
     */
    private String hosts;

    /**
     * 通过Http协议访问的主机地址，格式是 ip:port,ip1:port1.....
     */
    private String httpHosts;

    /**
     * es集群名称
     */
    private String clusterName;

    /**
     * ping超时时长
     */
    private String pingTimeout;

    public String getHosts() {
        return hosts;
    }

    public void setHosts(String hosts) {
        this.hosts = hosts;
    }

    public String getHttpHosts() {
        return httpHosts;
    }

    public void setHttpHosts(String httpHosts) {
        this.httpHosts = httpHosts;
    }

    public String getClusterName() {
        return clusterName;
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    public String getPingTimeout() {
        return pingTimeout;
    }

    public void setPingTimeout(String pingTimeout) {
        this.pingTimeout = pingTimeout;
    }

    /**
     * 通信客户端
     */
    private Map<String, TransportClient> clientMap = new ConcurrentHashMap<>();

    /**
     * 通过 TransportClient 这个接口，我们可以不启动节点就可以和 es 集群进行通信，它需要指定 es 集群中其中一台或多台机器的 p
     * 地址和端口,该节点不加入集群 设置 client.transport.sniff==true允许客户端嗅探集群中的机器列表加入当前节点的机器列表
     *
     * @return
     */
    private Client getClient() {
        TransportClient client = clientMap.get(clusterName);
        if (client == null) {
            String[] hostArr = StringUtils.split(hosts, ',');
            validate(hostArr);
            Settings settings = ImmutableSettings.settingsBuilder()
                    .put("cluster.name", clusterName)
                    .put("client.transport.ping_timeout", pingTimeout)
                    .put("client.transport.sniff", true)
                    .build();
            client = new TransportClient(settings);
            for (String host : hostArr) {
                String[] hostAddress = host.split(":");
                String ipAddress = hostAddress[0];
                String port = hostAddress[1];
                client.addTransportAddress(new InetSocketTransportAddress(ipAddress, Integer.parseInt(port)));
            }
            clientMap.put(clusterName, client);
        }
        return client;
    }

    /**
     * 校验elasticsearch hosts
     *
     * @param hosts
     */
    private void validate(String[] hosts) {
        if (hosts == null || hosts.length <= 0) {
            throw new ElasticException("elasticsearch hosts can not null");
        }
        for (String host : hosts) {
            if (StringUtils.isBlank(host)) {
                throw new ElasticException("elasticsearch hosts " + host + " can not null");
            }
            String[] hostPost = StringUtils.split(host, ':');
            if (hostPost.length != 2) {
                throw new ElasticException("elasticsearch hosts must be ip1:port1,ip2:port2...");
            }
        }
    }

    @Override
    public Result createIndex(EsIndex esIndex) throws Exception {
        CreateIndexResponse response;
        try {
            response = getClient().admin().indices().prepareCreate(esIndex.getIndexName()).execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return new Result(Result.SUCCESS, response.isAcknowledged());
    }

    @Override
    public Result deleteIndex(EsIndex esIndex) throws Exception {
        DeleteIndexResponse response;
        try {
            response = getClient().admin().indices().prepareDelete(esIndex.getIndexName()).execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return new Result(Result.SUCCESS, response.isAcknowledged());
    }

    @Override
    public Result createType(EsType esType) throws Exception {
        List<EsField> esFieldList = esType.getFields();
        if (CollectionUtils.isEmpty(esFieldList)) {
            throw new ElasticException("elasticsearch fields can not be null");
        }
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        builder.startObject(esType.getIndexType());
        builder.startObject("properties");
        buildEsFields(esFieldList, builder);
        builder.endObject();
        builder.endObject();
        builder.endObject();
        try {
            PutMappingRequest mappingRequest = Requests.putMappingRequest(esType.getIndexName()).type(esType.getIndexType()).source(builder);
            getClient().admin().indices().putMapping(mappingRequest).actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return Result.getSuccessResult(null);
    }

    private void buildEsFields(List<EsField> esFieldList, XContentBuilder builder) throws IOException {
        for (EsField esField : esFieldList) {
            builder.startObject(esField.getFieldName());
            //字段类型
            String fieldType = StringUtils.isBlank(esField.getFieldName()) ? "string" : esField.getType();
            builder.field("type", fieldType);
            if (!StringUtils.equals(fieldType, "completion")) {
                //是否存储
                String store = StringUtils.isBlank(esField.getStore()) ? "false" : esField.getStore();
                builder.field("store", store);
                //是否分词
                String index = StringUtils.isBlank(esField.getIndex()) ? "not_analyzed" : esField.getIndex();
                builder.field("index", index);
            } else {
                builder.field("payloads", "true");
            }
            //字段复制
            String copyTo = esField.getCopyTo();
            if (StringUtils.isNotBlank(copyTo)) {
                builder.field("copy_to", copyTo);
            }
            //空值时的赋值，该类型需要跟该字段type的类型一致
            Object nullValue = esField.getNullValue();
            if (nullValue != null) {
                builder.field("null_value", nullValue);
            }
            //多字段从属性
            List<EsField> fields = esField.getFields();
            if (CollectionUtils.isNotEmpty(fields)) {
                XContentBuilder xContentBuilder = XContentFactory.jsonBuilder();
                buildEsFields(fields, xContentBuilder);
            }
            //分词解析
            String indexAnalyzer = esField.getIndexAnalyzer();
            if (StringUtils.isNotBlank(indexAnalyzer)) {
                builder.field("index_analyzer", indexAnalyzer);
            }
            //搜索解析
            String searchAnalyzer = esField.getSearchAnalyzer();
            if (StringUtils.isNotBlank(searchAnalyzer)) {
                builder.field("search_analyzer", searchAnalyzer);
            }
            builder.endObject();
        }
    }

    @Override
    public Result deleteType(EsType esType) throws Exception {
        SearchResponse scrollResp;
        try {
            if (!checkTypesExists(esType).isExists()) {
                throw new ElasticException("elasticsearch index:" + esType.getIndexName() + " not exists type:" + esType.getIndexType());
            }
            scrollResp = getClient().prepareSearch(esType.getIndexName())
                    .setTypes(esType.getIndexType())
                    .setSearchType(SearchType.SCAN)
                    .setScroll(new TimeValue(60000))
                    .setSize(100).execute().actionGet();//100 hits per shard will be returned for each scroll
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        while (true) {
            try {
                scrollResp = getClient().prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(600000)).execute().actionGet();
                if (scrollResp.getHits().getHits().length == 0) {
                    break;
                }
                BulkRequestBuilder bulkRequest = getClient().prepareBulk();
                for (SearchHit searchHit : scrollResp.getHits().getHits()) {
                    String indexId = searchHit.getId();
                    DeleteRequest request = getClient().prepareDelete(esType.getIndexName(), esType.getIndexType(), indexId)
                            .setOperationThreaded(false)
                            .request();
                    bulkRequest.add(request);
                }
                bulkRequest.execute().actionGet();
            } catch (Exception e) {
                throw new ElasticException(e.getMessage());
            }
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result isExistsIndices(EsIndex esIndex) throws Exception {
        Result result;
        try {
            IndicesExistsResponse response = checkIndicesExists(esIndex);
            if (response.isExists()) {
                result = new Result(Result.SUCCESS, response.isExists());
            } else {
                result = new Result(Result.FAIL, response.isExists());
            }
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return result;
    }

    private IndicesExistsResponse checkIndicesExists(EsIndex esIndex) {
        IndicesExistsRequestBuilder indicesExistsRequestBuilder = getClient().admin().indices().prepareExists(esIndex.getIndexName());
        return indicesExistsRequestBuilder.execute().actionGet();
    }

    @Override
    public Result isExistsType(EsType esType) throws Exception {
        Result result;
        try {
            TypesExistsResponse response = checkTypesExists(esType);
            if (response.isExists()) {
                result = new Result(Result.SUCCESS, response.isExists());
            } else {
                result = new Result(Result.FAIL, response.isExists());
            }
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return result;
    }

    private TypesExistsResponse checkTypesExists(EsType esType) {
        List<String> indexNames = new ArrayList<>();
        indexNames.add(esType.getIndexName());
        String[] indexNameArr = new String[indexNames.size()];
        TypesExistsRequest request = new TypesExistsRequest(indexNames.toArray(indexNameArr), esType.getIndexType());
        return getClient().admin().indices().typesExists(request).actionGet();
    }

    @Override
    public Result deleteMapping(EsType esType) throws Exception {
        DeleteMappingResponse response;
        try {
            DeleteMappingRequestBuilder deleteMappingRequestBuilder = getClient().admin().indices().prepareDeleteMapping(esType.getIndexName());
            deleteMappingRequestBuilder.setType(esType.getIndexType());
            response = deleteMappingRequestBuilder.execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return new Result(Result.SUCCESS, response.isAcknowledged());
    }

    @Override
    public Result addDocument(EsDocument esDocument) throws Exception {
        String dataSource = JSONUtils.toJsonStr(esDocument.getData());
        IndexResponse response;
        try {
            response = getClient().prepareIndex(esDocument.getIndexName(), esDocument.getIndexType(), esDocument.getId())
                    .setSource(dataSource).execute().actionGet();
            addKeywordDocument(Collections.singletonList(esDocument));
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return new Result(Result.SUCCESS, response.isCreated());
    }

    @Override
    public Result batchAddDocument(List<EsDocument> esDocumentList) throws Exception {
        if (CollectionUtils.isEmpty(esDocumentList)) {
            throw new ElasticException("elasticsearch documents can not be null");
        }
        try {
            BulkRequestBuilder bulkRequest = getClient().prepareBulk();//批量请求
            for (EsDocument esDocument : esDocumentList) {
                IndexRequest request = getClient().prepareIndex(esDocument.getIndexName(), esDocument.getIndexType(), esDocument.getId())
                        .setSource(JSONUtils.toJsonStr(esDocument.getData()))
                        .request();
                bulkRequest.add(request);
            }
            bulkRequest.execute().actionGet();
            addKeywordDocument(esDocumentList);
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result addKeywordDocument(List<EsDocument> esDocumentList) throws Exception {
        Gson gson = new GsonBuilder().setPrettyPrinting().disableHtmlEscaping().create();
        JsonParser parser = new JsonParser();
        List<EsDocument> keywordDocumentList = new ArrayList<>();
        for (EsDocument esDocument : esDocumentList) {
            String suggestIndexName = esDocument.getSuggestIndexName();
            String suggestType = esDocument.getSuggestIndexType();
            List<String> suggestFields = esDocument.getSuggestFields();
            if (StringUtils.isNotBlank(suggestIndexName) && StringUtils.isNotBlank(suggestType) && CollectionUtils.isNotEmpty(suggestFields)) {
                JsonObject jsonObject = parser.parse(gson.toJson(esDocument.getData())).getAsJsonObject();
                for (String fieldName : suggestFields) {
                    JsonElement jsonElement = jsonObject.get(fieldName);
                    if (jsonElement != null) {
                        //原关键词
                        String kw = jsonElement.toString();
                        Set<String> pinyin = Pinyin4jUtil.getPinyinToLowerCase(kw);
                        Set<String> abbre = Pinyin4jUtil.getPinyinJianPinToLowerCase(kw);
                        EsDocument keywordEsDocument = new EsDocument();
                        keywordEsDocument.setIndexName(suggestIndexName);
                        keywordEsDocument.setIndexType(suggestType);
                        keywordEsDocument.setId(kw);
                        keywordEsDocument.setData(MapUtil.$m("kw", kw, "pinyin", pinyin, "abbre", abbre));
                        keywordDocumentList.add(keywordEsDocument);
                    }
                }
            }
        }
        try {
            batchAddDocument(keywordDocumentList);
        } catch (Exception ignore) {
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result updateDocument(EsDocument esDocument) throws Exception {
        String dataSource = JSONUtils.toJsonStr(esDocument.getData());
        try {
            UpdateRequest updateRequest = new UpdateRequest(esDocument.getIndexName(), esDocument.getIndexType(), esDocument.getId())
                    .doc(dataSource);
            getClient().update(updateRequest).get();
            addKeywordDocument(Collections.singletonList(esDocument));
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result batchUpdateDocument(List<EsDocument> esDocumentList) throws Exception {
        try {
            BulkRequestBuilder bulkRequest = getClient().prepareBulk();//批量请求
            for (EsDocument esDocument : esDocumentList) {
                UpdateRequest request = getClient().prepareUpdate(esDocument.getIndexName(), esDocument.getIndexType(), esDocument.getId())
                        .setDoc(JSONUtils.toJsonStr(esDocument.getData()))
                        .request();
                bulkRequest.add(request);
            }
            bulkRequest.execute().actionGet();
            addKeywordDocument(esDocumentList);
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result deleteDocument(EsDocument esDocument) throws Exception {
        try {
            getClient().prepareDelete(esDocument.getIndexName(), esDocument.getIndexType(), esDocument.getId())
                    .setOperationThreaded(false).execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result batchDeleteDocument(List<EsDocument> esDocumentList) throws Exception {
        try {
            BulkRequestBuilder bulkRequest = getClient().prepareBulk();
            for (EsDocument esDocument : esDocumentList) {
                DeleteRequest request = getClient().prepareDelete(esDocument.getIndexName(), esDocument.getIndexType(), esDocument.getId())
                        .setOperationThreaded(false)
                        .request();
                bulkRequest.add(request);
            }
            bulkRequest.execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result search(EsSearch esSearch) throws Exception {
        if (StringUtils.isBlank(esSearch.getIndex())) {
            throw new ElasticException("elasticsearch indexName can not be null");
        }
        if (CollectionUtils.isEmpty(esSearch.getTypes())) {
            throw new ElasticException("elasticsearch indexTypes can not be null");
        }

        List<String> indexTypes = esSearch.getTypes();
        String[] indexTypeArr = new String[indexTypes.size()];
        indexTypes.toArray(indexTypeArr);
        SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(esSearch.getIndex())
                .setTypes(indexTypeArr)
                .setSearchType(SearchType.DFS_QUERY_THEN_FETCH);

        List<String> fields = esSearch.getFields();
        QueryBuilder queryBuilder;//查询构造器
        if (CollectionUtils.isNotEmpty(fields) && !esSearch.isExactSearch()) {//多域查询
            String[] fieldArr = new String[fields.size()];
            fields.toArray(fieldArr);
            queryBuilder = QueryBuilders.multiMatchQuery(esSearch.getKeyWord(), fieldArr);
        } else if (CollectionUtils.isNotEmpty(fields) && esSearch.isExactSearch()) {
            queryBuilder = QueryBuilders.queryStringQuery(esSearch.getKeyWord());
            for (String field : fields) {
                ((QueryStringQueryBuilder) queryBuilder).field(field);
            }
        } else {
            queryBuilder = QueryBuilders.queryStringQuery(esSearch.getKeyWord());
        }

        //过滤查询
        FilterBuilder filterBuilder = buildFilterBuilder(esSearch.getMustCondition(), esSearch.getMustNotCondition(), esSearch.getShouldCondition(), esSearch.getRangeCondition());
        queryBuilder = QueryBuilders.filteredQuery(queryBuilder, filterBuilder);

        //分页查询
        int currentPage = (esSearch.getCurrentPage() - 1) < 0 ? 0 : (esSearch.getCurrentPage() - 1);
        int pageSize = esSearch.getPageSize();
        searchRequestBuilder.setFrom(currentPage).setSize(pageSize);

        //高亮查询
        List<EsHighlightField> highLighterFields = esSearch.getHighLighterFields();
        if (CollectionUtils.isNotEmpty(highLighterFields)) {
            for (EsHighlightField highLighterField : highLighterFields) {
                searchRequestBuilder.addHighlightedField(highLighterField.getName(),
                        highLighterField.getFragmentsSize(), highLighterField.getNumberOfFragments());
            }
            searchRequestBuilder.setHighlighterPreTags(esSearch.getHighLighterPreTag());
            searchRequestBuilder.setHighlighterPostTags(esSearch.getHighLighterPostTag());
        }

        //排序查询
        buildSortBuilder(esSearch, searchRequestBuilder);

        SearchResponse response;
        try {
            response = searchRequestBuilder.setQuery(queryBuilder).setExplain(true).execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        SearchHits searchHits = response.getHits();
        EsSearchResult esResult = searchHits2EsResult(searchHits);
        return new Result(Result.SUCCESS, esResult);
    }

    @Override
    public Result addJDBCRiverConfig(EsJDBCRiverConfig jdbcRiverConfig, String riverName) throws Exception {
        if (StringUtils.isBlank(riverName)) {
            throw new ElasticException("elasticsearch riverName can not be null");
        }
        String[] hostArr = StringUtils.split(httpHosts, ',');
        validate(hostArr);
        Gson gson = new GsonBuilder().create();
        String jdbcRiverConfigString = gson.toJson(jdbcRiverConfig);
        for (String httpHost : hostArr) {
            String result = HttpUtil.sendHttpRequest(HttpUtil.HttpMethodType.PUT, "http://" + httpHost + "/_river/" + riverName + "/_meta",
                    jdbcRiverConfigString, "UTF-8", false);
            JsonParser parser = new JsonParser();
            JsonObject jsonObject = parser.parse(result).getAsJsonObject();
            if (!Boolean.parseBoolean(jsonObject.get("created").getAsString())) {
                throw new ElasticException("elasticsearch add jdbc river failure:" + result);
            }
        }
        return Result.getSuccessResult(null);
    }

    @Override
    public Result search(EsAggregation esAggregation) throws Exception {
        if (StringUtils.isBlank(esAggregation.getIndex())) {
            throw new ElasticException("elasticsearch indexName can not be null");
        }
        if (CollectionUtils.isEmpty(esAggregation.getTypes())) {
            throw new ElasticException("elasticsearch indexTypes can not be null");
        }
        //分组查询
        List<String> indexTypes = esAggregation.getTypes();
        String[] indexTypeArr = new String[indexTypes.size()];
        indexTypes.toArray(indexTypeArr);
        SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(esAggregation.getIndex())
                .setTypes(indexTypeArr)
                .setSearchType(SearchType.COUNT);
        List<AggregationBuilder<?>> aggregationBuilderList = new ArrayList<>();
        buildAggregation(esAggregation, aggregationBuilderList);
        for (AggregationBuilder<?> aggregationBuilder : aggregationBuilderList) {
            searchRequestBuilder.addAggregation(aggregationBuilder);
        }

        //过滤查询
        FilterBuilder filterBuilder = buildFilterBuilder(esAggregation.getMustCondition(), esAggregation.getMustNotCondition(), esAggregation.getShouldCondition(), esAggregation.getRangeCondition());
        QueryBuilder queryBuilder = QueryBuilders.filteredQuery(null, filterBuilder);
        searchRequestBuilder.setQuery(queryBuilder);

        SearchResponse response;
        try {
            response = searchRequestBuilder.setExplain(true).execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        Map<String, Aggregation> aggregationMap = response.getAggregations().asMap();
        Map<String, List<Map<String, Object>>> aggregationDataMap = buildAggregationDataMap(aggregationMap);
        return Result.getSuccessResult(aggregationDataMap);
    }

    private Map<String, List<Map<String, Object>>> buildAggregationDataMap(Map<String, Aggregation> aggregationMap) {
        Map<String, List<Map<String, Object>>> aggregationDataMap = new HashMap<>();
        List<Map<String, Object>> bucketDataMapList = new ArrayList<>();
        for (String aggregationKey : aggregationMap.keySet()) {
            if (aggregationKey.contains("_null")) {
                continue;
            }
            Aggregation aggregation = aggregationMap.get(aggregationKey);
            if (aggregation != null && aggregation instanceof MultiBucketsAggregation) {
                MultiBucketsAggregation multiBucketsAggregation = (MultiBucketsAggregation) aggregation;
                for (MultiBucketsAggregation.Bucket bucket : multiBucketsAggregation.getBuckets()) {
                    Map<String, Object> bucketDataMap = new HashMap<>();
                    bucketDataMap.put("key_as_string", bucket.getKeyAsText().string());
                    bucketDataMap.put("doc_count", bucket.getDocCount());
                    Map<String, Aggregation> subAggregationMap = bucket.getAggregations().asMap();
                    Map<String, List<Map<String, Object>>> subAggregationDataMap = buildAggregationDataMap(subAggregationMap);
                    if (MapUtils.isNotEmpty(subAggregationDataMap)) {
                        bucketDataMap.put("buckets", subAggregationDataMap);
                    }
                    bucketDataMapList.add(bucketDataMap);
                }
            }
            Aggregation missingAggregation = aggregationMap.get(aggregationKey + "_null");
            if (missingAggregation != null && missingAggregation instanceof SingleBucketAggregation) {
                SingleBucketAggregation singleBucketAggregation = (SingleBucketAggregation) missingAggregation;
                if (singleBucketAggregation.getDocCount() > 0) {
                    Map<String, Object> singleBucketDataMap = new HashMap<>();
                    singleBucketDataMap.put("key_as_string", "NA");
                    singleBucketDataMap.put("doc_count", singleBucketAggregation.getDocCount());
                    //获取bucket
                    Map<String, Aggregation> singleSubAggregationMap = singleBucketAggregation.getAggregations().asMap();
                    Map<String, List<Map<String, Object>>> singleSubAggregationDataMap = buildAggregationDataMap(singleSubAggregationMap);
                    if (MapUtils.isNotEmpty(singleSubAggregationDataMap)) {
                        singleBucketDataMap.put("buckets", singleSubAggregationDataMap);
                    }
                    bucketDataMapList.add(singleBucketDataMap);
                }
            }
            aggregationDataMap.put(aggregationKey, bucketDataMapList);
        }
        return aggregationDataMap;
    }

    private void buildAggregation(EsAggregation esAggregation, List<AggregationBuilder<?>> aggregationBuilderList) {
        AggregationBuilder<?> aggregationBuilder;
        AggregationBuilder<?> missingAggregationBuilder = null;
        if (StringUtils.equals(esAggregation.getFieldType(), "date")) {
            EsDateHistogram esDateHistogram = esAggregation.getEsDateHistogram();
            aggregationBuilder = AggregationBuilders
                    .dateHistogram(esAggregation.getAggregationName())
                    .field(esAggregation.getFieldName())
                    .minDocCount(Long.parseLong(esDateHistogram.getMinDocCount()))
                    .format(esDateHistogram.getFormat())
                    .timeZone(TimeZone.getDefault().getID());
            if ("SECOND".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.SECOND);
            } else if ("MINUTE".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.MINUTE);
            } else if ("HOUR".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.HOUR);
            } else if ("DAY".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.DAY);
            } else if ("WEEK".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.WEEK);
            } else if ("MONTH".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.MONTH);
            } else if ("QUARTER".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.QUARTER);
            } else if ("YEAR".equalsIgnoreCase(esDateHistogram.getInterval())) {
                ((DateHistogramBuilder) aggregationBuilder).interval(DateHistogram.Interval.YEAR);
            }
        } else {
            aggregationBuilder = AggregationBuilders.terms(esAggregation.getAggregationName()).field(esAggregation.getFieldName()).size(esAggregation.getSize() != null ? esAggregation.getSize() : 10);
            if (esAggregation.getMissing()) {
                missingAggregationBuilder = AggregationBuilders.missing(esAggregation.getAggregationName() + "_null").field(esAggregation.getFieldName());
            }
        }

        EsAggregation subAggregations = esAggregation.getSubAggregation();
        if (subAggregations == null) {
            aggregationBuilderList.add(aggregationBuilder);
            if (missingAggregationBuilder != null) {
                aggregationBuilderList.add(missingAggregationBuilder);
            }
            return;
        }
        List<AggregationBuilder<?>> subAggregationBuilderList = new ArrayList<>();
        buildAggregation(subAggregations, subAggregationBuilderList);
        for (AggregationBuilder<?> subAggregationBuilder : subAggregationBuilderList) {
            aggregationBuilder.subAggregation(subAggregationBuilder);
            if (missingAggregationBuilder != null) {
                missingAggregationBuilder.subAggregation(subAggregationBuilder);
            }
        }
        aggregationBuilderList.add(aggregationBuilder);
        if (missingAggregationBuilder != null) {
            aggregationBuilderList.add(missingAggregationBuilder);
        }
    }

    /**
     * 构建排序查询
     *
     * @param esSearch
     * @param searchRequestBuilder
     */
    private void buildSortBuilder(EsSearch esSearch, SearchRequestBuilder searchRequestBuilder) {
        if (StringUtils.isNotBlank(esSearch.getSortByParamNames()) && StringUtils.isNotBlank(esSearch.getSortByParamTypes())
                && esSearch.getSortByParamNames().length() == esSearch.getSortByParamTypes().length()) {
            String[] sortByParamNames = StringUtils.split(esSearch.getSortByParamNames(), ',');
            String[] sortByParamTypes = StringUtils.split(esSearch.getSortByParamTypes(), ',');
            for (int i = 0; i <= sortByParamNames.length; i++) {
                String sortByParamName = sortByParamNames[i];
                String sortByParamType = sortByParamTypes[i];
                if (StringUtils.equals(sortByParamType, "asc") || StringUtils.equals(sortByParamType, "ASC"))
                    searchRequestBuilder.addSort(sortByParamName, SortOrder.ASC);
                else if (StringUtils.equals(sortByParamType, "desc") || StringUtils.equals(sortByParamType, "DESC")) {
                    searchRequestBuilder.addSort(sortByParamName, SortOrder.DESC);
                }
            }
        }
    }

    /**
     * 构造过滤器
     *
     * @param mustCondition
     * @param mustNotCondition
     * @param shouldCondition
     * @param rangeCondition
     * @return
     */
    private FilterBuilder buildFilterBuilder(String mustCondition, String mustNotCondition, String shouldCondition, String rangeCondition) {
        BoolFilterBuilder boolFilterBuilder = FilterBuilders.boolFilter();
        if (StringUtils.isNotBlank(mustCondition)) {
            String[] conditions = mustCondition.split("&");
            for (String condition : conditions) {
                String[] conditionArr = condition.split(":");
                if ("NA".equalsIgnoreCase(conditionArr[1])) {
                    boolFilterBuilder.must(FilterBuilders.missingFilter(conditionArr[0]));
                } else {
                    boolFilterBuilder.must(FilterBuilders.termFilter(conditionArr[0], conditionArr[1]));
                }
            }
        }
        if (StringUtils.isNotBlank(mustNotCondition)) {
            String[] conditions = mustNotCondition.split("&");
            for (String condition : conditions) {
                String[] conditionArr = condition.split(":");
                if ("NA".equalsIgnoreCase(conditionArr[1])) {
                    boolFilterBuilder.mustNot(FilterBuilders.missingFilter(conditionArr[0]));
                } else {
                    boolFilterBuilder.mustNot(FilterBuilders.termFilter(conditionArr[0], conditionArr[1]));
                }
            }
        }
        if (StringUtils.isNotBlank(shouldCondition)) {
            String[] conditions = shouldCondition.split("&");
            for (String condition : conditions) {
                String[] conditionArr = condition.split(":");
                String conditionFieldName = conditionArr[0];
                String conditionFieldValue = conditionArr[1];
                if (StringUtils.isNotBlank(conditionFieldName) && StringUtils.isNotBlank(conditionFieldValue)) {
                    String[] conditionFieldValues = conditionFieldValue.split(",");
                    List<String> existsFileValueList = new ArrayList<>();
                    for (String fieldValue : conditionFieldValues) {
                        if ("NA".equalsIgnoreCase(fieldValue)) {
                            boolFilterBuilder.should(FilterBuilders.missingFilter(conditionArr[0]));
                        } else {
                            existsFileValueList.add(fieldValue);
                        }
                    }
                    if (CollectionUtils.isNotEmpty(existsFileValueList)) {
                        boolFilterBuilder.should(FilterBuilders.inFilter(conditionFieldName, existsFileValueList.toArray()));
                    }
                }
            }
        }
        if (StringUtils.isNotBlank(rangeCondition)) {
            Map<String, List<String>> rangeMap = SearchConditionUtil.getRangeByStr(rangeCondition);
            for (String rangeExp : rangeMap.keySet()) {
                List<String> rangeList = rangeMap.get(rangeExp);
                RangeFilterBuilder rangeFilterBuilder = FilterBuilders.rangeFilter(rangeList.get(0));
                rangeFilterBuilder.timeZone(TimeZone.getDefault().getID());
                if (rangeList.get(1).contains(">")) {
                    if (SearchConditionUtil.checkValidDate(rangeList.get(2))) {
                        rangeFilterBuilder.from(DateUtil.toDate(rangeList.get(2)));
                    } else {
                        rangeFilterBuilder.from(rangeList.get(2));
                    }
                    rangeFilterBuilder.includeLower(rangeList.get(1).contains("="));
                } else if (rangeList.get(1).contains("<")) {
                    if (SearchConditionUtil.checkValidDate(rangeList.get(2))) {
                        rangeFilterBuilder.to(SearchConditionUtil.getDayLastTime(DateUtil.toDate(rangeList.get(2))));
                    } else {
                        rangeFilterBuilder.to(rangeList.get(2));
                    }
                    rangeFilterBuilder.includeUpper(rangeList.get(1).contains("="));
                }
                boolFilterBuilder.must(rangeFilterBuilder);
            }
        }
        FilterBuilder filterBuilder = null;
        if (boolFilterBuilder.hasClauses()) {
            filterBuilder = FilterBuilders.andFilter(boolFilterBuilder);
        }
        return filterBuilder;
    }

    /**
     * 查询命中结果转换
     *
     * @param searchHits
     * @return
     */
    private EsSearchResult searchHits2EsResult(SearchHits searchHits) {
        EsSearchResult esSearchResult = new EsSearchResult();
        esSearchResult.setMaxScore(searchHits.getMaxScore());
        esSearchResult.setTotalHits(searchHits.getTotalHits());
        SearchHit[] searchHitArr = searchHits.getHits();
        if (searchHitArr != null && searchHitArr.length > 0) {
            EsSearchHit[] esSearchHits = new EsSearchHit[searchHitArr.length];
            List<EsSearchHit> esSearchHitList = new ArrayList<>();
            for (SearchHit searchHit : searchHitArr) {
                EsSearchHit esSearchHit = new EsSearchHit();
                esSearchHit.setId(searchHit.getId());
                esSearchHit.setIndex(searchHit.getIndex());
                esSearchHit.setScore(searchHit.getScore());
                esSearchHit.setSourceAsMap(searchHit.getSource());
                esSearchHit.setSourceAsString(searchHit.getSourceAsString());
                esSearchHit.setSourceEmpty(searchHit.isSourceEmpty());
                esSearchHit.setType(searchHit.getType());
                esSearchHit.setVersion(searchHit.getVersion());
                //高亮
                Map<String, HighlightField> highlightFieldMap = searchHit.getHighlightFields();
                if (MapUtils.isNotEmpty(highlightFieldMap)) {
                    Map<String, EsHighlightField> esHighlightFieldMap = new HashMap<>();
                    Set<String> highlightFields = highlightFieldMap.keySet();
                    for (String highlightField : highlightFields) {
                        HighlightField hlField = highlightFieldMap.get(highlightField);
                        EsHighlightField esHighlightField = new EsHighlightField();
                        esHighlightField.setName(hlField.getName());
                        if (hlField.getFragments() != null && hlField.getFragments().length > 0) {
                            String[] fragments = new String[hlField.getFragments().length];
                            for (int i = 0; i < hlField.getFragments().length; i++) {
                                fragments[i] = hlField.getFragments()[i].toString();
                            }
                            esHighlightField.setFragments(fragments);
                        }
                        esHighlightFieldMap.put(highlightField, esHighlightField);
                    }
                    esSearchHit.setHighlightFields(esHighlightFieldMap);
                }
                esSearchHitList.add(esSearchHit);
            }
            esSearchResult.setHits(esSearchHitList.toArray(esSearchHits));
        }
        return esSearchResult;
    }


    /**
     * Suggest查询命中结果转换,该方法尽量减少返回值,有待优化
     *
     * @param searchHits
     * @return
     */
    private List<EsSearchHit> suggestHits2EsResult(SearchHits searchHits) {
        List<EsSearchHit> esSearchHitList = new ArrayList<>();
        SearchHit[] searchHitArr = searchHits.getHits();
        if (searchHitArr != null && searchHitArr.length > 0) {
            for (SearchHit searchHit : searchHitArr) {
                EsSearchHit esSearchHit = new EsSearchHit();
                esSearchHit.setId(searchHit.getId());
                esSearchHit.setScore(searchHit.getScore());
                esSearchHit.setVersion(searchHit.getVersion());
                esSearchHitList.add(esSearchHit);
            }
        }
        return esSearchHitList;
    }

    @Override
    public Result searchCompletionSuggest(EsSearch esSearch) {
        CompletionSuggestionBuilder suggestionsBuilder = new CompletionSuggestionBuilder("completion");
        suggestionsBuilder.text(esSearch.getKeyWord());
        suggestionsBuilder.field("suggest");
        suggestionsBuilder.size(10);
        suggestionsBuilder.analyzer("ik_analyzer");
        SuggestResponse resp = getClient().prepareSuggest(esSearch.getIndex())
                .addSuggestion(suggestionsBuilder).execute().actionGet();
        List<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> list = resp.getSuggest().getSuggestion("completion").getEntries();
        List<String> suggests = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(list)) {
            for (Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option> e : list) {
                for (Suggest.Suggestion.Entry.Option option : e) {
                    suggests.add(option.getText().toString());
                }
            }
        }
        return Result.getSuccessResult(suggests);
    }

    @Override
    public Result searchKeywordSuggest(EsSearch esSearch) {
        if (StringUtils.isBlank(esSearch.getIndex())) {
            throw new ElasticException("elasticsearch indexName can not be null");
        }
        if (CollectionUtils.isEmpty(esSearch.getTypes())) {
            throw new ElasticException("elasticsearch indexTypes can not be null");
        }

        List<String> indexTypes = esSearch.getTypes();
        String[] indexTypeArr = new String[indexTypes.size()];
        indexTypes.toArray(indexTypeArr);
        SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(esSearch.getIndex())
                .setTypes(indexTypeArr)
                .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                .setSize(10)
                .setVersion(Boolean.TRUE);
        List<String> fields = esSearch.getFields();

        QueryBuilder queryBuilder;//查询构造器
        if (CollectionUtils.isEmpty(fields)) {
            queryBuilder = QueryBuilders.matchQuery("suggest", esSearch.getKeyWord()).operator(MatchQueryBuilder.Operator.AND);
        } else {
            String[] fieldArr = new String[fields.size()];
            fields.toArray(fieldArr);
            queryBuilder = QueryBuilders.multiMatchQuery(esSearch.getKeyWord(), fieldArr);
        }
        //高亮查询
        List<EsHighlightField> highLighterFields = esSearch.getHighLighterFields();
        if (CollectionUtils.isNotEmpty(highLighterFields)) {
            for (EsHighlightField highLighterField : highLighterFields) {
                searchRequestBuilder.addHighlightedField(highLighterField.getName(),
                        highLighterField.getFragmentsSize(), highLighterField.getNumberOfFragments());
            }
            searchRequestBuilder.setHighlighterPreTags(esSearch.getHighLighterPreTag());
            searchRequestBuilder.setHighlighterPostTags(esSearch.getHighLighterPostTag());
        }

        //排序查询
        buildSortBuilder(esSearch, searchRequestBuilder);

        SearchResponse response;
        try {
            response = searchRequestBuilder.setQuery(queryBuilder).setExplain(true).execute().actionGet();
        } catch (Exception e) {
            throw new ElasticException(e.getMessage());
        }
        SearchHits searchHits = response.getHits();
        List<EsSearchHit> esSearchHitList = suggestHits2EsResult(searchHits);
        return new Result(Result.SUCCESS, esSearchHitList);
    }

    @Override
    public void destroy() throws Exception {
        for (String clusterName : clientMap.keySet()) {
            TransportClient client = clientMap.get(clusterName);
            if (client != null) {
                logger.error("elasticsearch[" + clusterName + "] current client is destroy...");
                client.close();
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    for (String clusterName : clientMap.keySet()) {
                        TransportClient client = clientMap.get(clusterName);
                        if (client != null) {
                            logger.error("elasticsearch[" + clusterName + "] current client is closed...");
                            client.close();
                        }
                    }
                } catch (Throwable ignore) {
                }
            }
        }));
    }
}
